package org.ethereum.net.dpos;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;

import org.ethereum.config.SystemProperties;
import org.ethereum.core.Block;
import org.ethereum.core.Blockchain;
import org.ethereum.facade.SyncStatus;
import org.ethereum.listener.CompositeEthereumListener;
import org.ethereum.listener.EthereumListenerAdapter;
import org.ethereum.net.rlpx.Node;
import org.ethereum.sync.SyncManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.spongycastle.util.encoders.Hex;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class ElectionNode implements Runnable {

	private static final Logger logger = LoggerFactory.getLogger("dpos");

	// the status of election organization.
	public enum Status {
		LOOKING, SELECTED;
	}
	
	Status state = Status.LOOKING;
	
	// the election members which supposed pre-configure over the network.
	final ConcurrentHashMap<String, Node> electionMemebers = new ConcurrentHashMap<String, Node>(4);

	// let give a limitation of candidates.
	private static final int CANDIDATE_MAXMEMBERS = 32;

	// the candidates' table of all possible nodes after proving by PoW question and
	// even more filtering conditions.
	final List<Node> candidatesTable;

	final ScheduledExecutorService votingScheduler;
	
	// the current voting pool.
	VoteInfo currentVotingPool;

	// the elected president list after latest block which is refreshed after a best
	// block sync and verified.
	final ConcurrentHashMap<Long, VotedPresidentBroadcast> electedPresidents = 
			new ConcurrentHashMap<Long, VotedPresidentBroadcast>(8);

	private DPOSHandler handler;

	private Blockchain blockchain;
	
	private SyncManager syncManager;

	private byte[] currentNodeId;

	private byte[] electionLeaderId;

	private final boolean isLeaderNode;
	
	private boolean startElection = false;
	
	private int votingInterval;

	@Autowired
	public ElectionNode(Blockchain blockchain, SyncManager syncManager, DPOSHandler handler, CompositeEthereumListener listener) {
		this.blockchain = blockchain;
		this.handler = handler;
		this.handler.setElectionNode(this);
		this.syncManager = syncManager;
		this.currentNodeId = Hex.decode(SystemProperties.getDefault().getGeneratedNodeId());
		// election nodes are configured at this moment.
		this.electionLeaderId = Hex.decode(SystemProperties.getDefault().dposElectionNodeInfo());
		logger.info("Election leader node info: {}", SystemProperties.getDefault().dposElectionNodeInfo());
		logger.debug("Current node info: {}", SystemProperties.getDefault().getGeneratedNodeId());
		listener.addListener(new EthereumListenerAdapter() {
			@Override
			public void onSyncDone(SyncState state) {
				startElection = true;
			}
			
		    @Override
		    public void onSyncedNewBlock(Block blocks) {
		    		if (!SystemProperties.getDefault().isDPOSEnabled()) {
		    			return;
		    		}
		    		if (!isLeaderNode()) {
		    			List<Long> roundList = new ArrayList<Long>();
		    			roundList.add(blocks.getHeader().getRound());
		    			ConfirmedSyncMessage confirmedMsg = new ConfirmedSyncMessage(roundList, getCurrentNodeId());
		    			handler.sendConfirmedSyncToElectionNode(confirmedMsg);
		    		}
		    		if (isLeaderNode()) {
		    			if (logger.isDebugEnabled())
		    				logger.debug("election leader node received a synced block {}", blocks.getHeader().getNumber());
		    		}
		    }
		});
		if (Arrays.equals(this.currentNodeId, this.electionLeaderId)) {
			logger.info("I am so proud to be the electing leader!");
			isLeaderNode = true;
			votingScheduler = Executors.newSingleThreadScheduledExecutor();
			candidatesTable = new ArrayList<Node>(CANDIDATE_MAXMEMBERS);
			votingInterval = SystemProperties.getDefault().dopsVotingInterval();
			votingScheduler.scheduleAtFixedRate(this, votingInterval, votingInterval, TimeUnit.SECONDS);
		} else {
			votingScheduler = null;
			candidatesTable = null;
			isLeaderNode = false;
		}
	}

	public void addElectionNode(String nodeId) {

	}

	public void addCandidateNode(RegisterCandidateRequest request) {
		if (!this.isLeaderNode) {
			logger.warn("I am not leader node. reject any registering request! {}", request.toString());
			return;
		}
		if (candidatesTable.size() > CANDIDATE_MAXMEMBERS) {
			logger.info("Candidates' table is full({})!", candidatesTable.size());
			return;
		}
		Node n = handler.getActivePeerNode(request.getCandidateId());
		if (n == null) {
			// now this is fake node due to the peer connection did not establish.
			logger.info("Failed to add candidate {} due to active peer is not existing.", request.toString());
			return;
		}
		boolean flag = false;
		for (Node e : candidatesTable) {
			if (Arrays.equals(e.getId(), n.getId())) {
				flag = true;
			}
		}
		if (!flag) {
			candidatesTable.add(n);
			logger.info("Candidates updated as: " + candidatesTable.toString());
		}
	}

	public byte[] getElectionLeader() {
		return electionLeaderId;
	}
	
	public byte[] getCurrentNodeId() {
		return currentNodeId;
	}

	public boolean isLeaderNode() {
		return isLeaderNode;
	}
	
	public List<Node> getCandidatesTable() {
		return candidatesTable;
	}

	public ConcurrentHashMap<String, Node> getElectionMemebers() {
		return electionMemebers;
	}

	@PostConstruct
	public void start() {
		if (!SystemProperties.getDefault().isDPOSEnabled()) {
			return;
		}
	}
	
	@PreDestroy
	public void stop() {
		if (this.isLeaderNode) {
			votingScheduler.shutdown();
		}
	}

	@Override
	public void run() {
		if (!startElection) {
			logger.info("waiting for blocks syncing completed at beginning.");
			return;
		}
		if (!isLeaderNode) {
			return;
		}
		if (candidatesTable.isEmpty()) {
			return;
		}
		try {
			// check the best block whether is synchronized or not.
			if (syncManager != null) {
				SyncStatus syncStatus = syncManager.getSyncStatus();
				if (syncStatus.getBlockBestKnown() > syncStatus.getBlockLastImported()) {
					logger.info("waiting for blocks syncing completed. {}", syncStatus.toString());
					return;
				}
			}
			
			// start voting...
			voteForNewPresident();

			// check keep-alive.
			List<byte[]> inactiveList = new ArrayList<byte[]>();
			for (Node n : candidatesTable) {
				if (handler.isActivePeer(n)) {
					// TODO: double check?
				} else {
					inactiveList.add(n.getId());
				}
			}
			inactiveList.stream().forEach(t -> {
				for (Node f : candidatesTable) {
					if (Arrays.equals(f.getId(), t)) {
						logger.info("Removed candidate from table due to inactive. node info: {}", f.toString());
						candidatesTable.remove(f);
						break;
					}
				}
			});
		} catch (Exception e) {
			logger.warn("voteForNewPresident error occurred with info: " + this.currentVotingPool, e);
		}
	}

	/**
	 * the event of president election is fired in every configurable second.
	 */
	private synchronized void voteForNewPresident() {
		long round = 1;
		// copy all candidates' table.
		List<Node> currCandidates = new ArrayList<Node>(this.candidatesTable);
		if (this.currentVotingPool != null) {
			// to make sure the health candidate pool of next voting, we'd better to remove the unconfirmed node
			// due to any possible issue including blocks in syncing, network unstability and etc.
			round = this.currentVotingPool.round + 1;
			List<String> unconfirmedNode = new ArrayList<String>();
			for (Map.Entry<String, AtomicInteger> entry : this.currentVotingPool.confirmedPool.entrySet()) {
				if (entry.getValue().get() == 0) {
					unconfirmedNode.add(entry.getKey());
				}
			}
			if (!unconfirmedNode.isEmpty()) {
				logger.info("Unconfirmed sync block candidates {}", unconfirmedNode.toString());
			}
			// let's remove the unconfirmed candidates for next round.
			for (String nId : unconfirmedNode) {
				byte[] nodeId = Hex.decode(nId);
				for (Node n : currCandidates) {
					if (Arrays.equals(nodeId, n.getId())) {
						currCandidates.remove(n);
						break;
					}
				}
			}
		} else {
			// query the round number from last block.
			round = blockchain.getBestBlock().getHeader().getRound() + 1;
		}
		if (currCandidates.isEmpty()) {
			// no any qualified candidate. set for next round.
			this.currentVotingPool = null;
			logger.warn("no any candidate confirmed the block synced in this round {}, revote again!", round);
			return ;
		}
		if (round < 1) {
			round = 1;
		}
		
		this.currentVotingPool = new VoteInfo(currCandidates, round);
		VotePresidentRequest vrequest = new VotePresidentRequest(round, this.currentVotingPool.getCandicatesIndex(),
				this.electionLeaderId);
		// broadcast this vote request to all nodes.
		if (handler.sendVoteRequest(vrequest)) {
			if (logger.isDebugEnabled())
				logger.debug("Voting the presidents: {}", this.currentVotingPool.toString());
			try {
				// wait for voting result in the gap of interval/2 ms which is applicable.
				Thread.sleep(this.votingInterval * 1000 / 2);
			} catch (InterruptedException e) {
			}
			
			// new request for packaging the block to remote peer.
			Node selectNode = this.currentVotingPool.maxTicket();
			PackageRequest prequest = new PackageRequest(this.currentVotingPool.round, selectNode,
					this.electionLeaderId);
			if (handler.sendPackageRequest(prequest)) {
				// all candidates must sending the confirmed sync message after mining new block.
				// this is important to make sure who are the best candidates in the next round.
				logger.info("Voted info: {}, Node info: {}", 
						new Object[] {this.currentVotingPool.toString(), selectNode.toString()});
			} else {
				this.candidatesTable.remove(selectNode);
				this.currentVotingPool = null;
				logger.warn("Failed to package the block from president({}) with {} round, revote again!",
						new Object[] { Hex.toHexString(prequest.getPresidentId()), prequest.getRound() });
			}
		} else {
			if (logger.isDebugEnabled())
				logger.debug("Discard this round during none of candidate's connection existing: {}",
						this.currentVotingPool.toString());
			this.currentVotingPool = null;
		}
	}

	public static class VoteInfo {
		boolean isClosed = false;
		final long round;
		final Map<String, AtomicInteger> confirmedPool = new ConcurrentHashMap<String, AtomicInteger>();
		final Map<Byte, AtomicInteger> votingPool = new ConcurrentHashMap<Byte, AtomicInteger>();
		final Map<Byte, Node> votingCandidcates = new ConcurrentHashMap<Byte, Node>();

		public VoteInfo(List<Node> candiSet, long round) {
			if (candiSet.size() >= 32) {
				throw new IllegalArgumentException("candicates of each round must be less then from 32");
			}
			byte i = 0;
			for (Node c : candiSet) {
				this.votingCandidcates.put(i, c);
				this.votingPool.put(i, new AtomicInteger(0));
				this.confirmedPool.put(Hex.toHexString(c.getId()), new AtomicInteger(0));
				i++;
			}
			if (round < 1) {
				throw new IllegalArgumentException("round must be greater then from 1!");
			}
			this.round = round;
		}

		public void voteFor(byte candicateIndex) {
			if (isClosed) {
				return;
			}
			if (votingPool.containsKey(candicateIndex)) {
				votingPool.get(candicateIndex).incrementAndGet();
				if (logger.isDebugEnabled()) {
					logger.debug("Just voted for president id: {}, total tickets: {} in {} round.",
							new Object[] { candicateIndex, votingPool.get(candicateIndex).get(), this.round });
				}
			}
		}
		
		/**
		 * confirmed by all candidates.
		 * @param rounds
		 * @param nodeId
		 */
		public void confirmSync(List<Long> rounds, byte[] nodeId) {
			for (long r : rounds) {
				if (r == this.round) {
					String k = Hex.toHexString(nodeId);
					if (confirmedPool.containsKey(k)) {
						confirmedPool.get(k).incrementAndGet();
						if (logger.isDebugEnabled()) {
							logger.debug("Just confirmed syncing completed {} in {} round.",
									new Object[] { Hex.toHexString(nodeId), this.round });
						}
					} else {
						// allows to be joined again.
						confirmedPool.put(k, new AtomicInteger(1));
					}
					break;
				}
			}
		}
		
		/**
		 * the president node failed to complete the task.
		 * @param pnodeId
		 */
		public void confirmSyncFailed(byte[] pnodeId) {
			for (Node n : votingCandidcates.values()) {
				if (!Arrays.equals(n.getId(), pnodeId)) {
					confirmedPool.put(Hex.toHexString(n.getId()), new AtomicInteger(1));
				}
			}
		}

		public byte[] getCandicatesIndex() {
			int i = 0;
			byte[] indexes = new byte[votingPool.size()];
			for (Byte key : votingPool.keySet()) {
				indexes[i++] = key;
			}
			return indexes;
		}

		public Node maxTicket() {
			isClosed = true;
			if (votingPool.size() <= 1) {
				return votingCandidcates.get((byte) 0);
			}
			Byte index = votingPool.entrySet().stream().max(new Comparator<Entry<Byte, AtomicInteger>>() {
				public int compare(Entry<Byte, AtomicInteger> o1, Entry<Byte, AtomicInteger> o2) {
					return o1.getValue().get() - o2.getValue().get();
				}
			}).get().getKey();
			return votingCandidcates.get(index);
		}

		public Node minTicket() {
			isClosed = true;
			if (votingPool.size() <= 1) {
				return votingCandidcates.get((byte) 0);
			}
			Byte index = votingPool.entrySet().stream().max(new Comparator<Entry<Byte, AtomicInteger>>() {
				public int compare(Entry<Byte, AtomicInteger> o1, Entry<Byte, AtomicInteger> o2) {
					return o2.getValue().get() - o1.getValue().get();
				}
			}).get().getKey();
			return votingCandidcates.get(index);
		}

		public Node meanTicket() {
			isClosed = true;
			// TODO:
			Byte index = votingPool.keySet().iterator().next();
			return votingCandidcates.get(index);
		}

		public String toString() {
			return "{round: " + round + ", votingPool: " + votingPool.toString() + ", confirmedPool: " + confirmedPool.toString() + " }";
		}
	}

}
